/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.gaia.translation.types;



import com.bff.gaia.unified.runners.gaia.translation.wrappers.DataInputViewWrapper;

import com.bff.gaia.unified.runners.gaia.translation.wrappers.DataOutputViewWrapper;

import com.bff.gaia.unified.sdk.coders.Coder;

import com.bff.gaia.unified.sdk.coders.CoderException;

import com.bff.gaia.unified.sdk.util.CoderUtils;

import com.bff.gaia.unified.vendor.guava.com.google.common.base.Preconditions;

import com.bff.gaia.api.common.typeutils.TypeSerializer;

import com.bff.gaia.api.common.typeutils.TypeSerializerConfigSnapshot;

import com.bff.gaia.api.common.typeutils.TypeSerializerSchemaCompatibility;

import com.bff.gaia.api.common.typeutils.TypeSerializerSnapshot;

import com.bff.gaia.core.io.VersionedIOReadableWritable;

import com.bff.gaia.core.memory.DataInputView;

import com.bff.gaia.core.memory.DataOutputView;



import java.io.EOFException;

import java.io.IOException;



/**

 * Gaia {@link com.bff.gaia.api.common.typeutils.TypeSerializer} for Unified {@link

 * Coder Coders}.

 */

public class CoderTypeSerializer<T> extends TypeSerializer<T> {



  private Coder<T> coder;



  public CoderTypeSerializer(Coder<T> coder) {

    Preconditions.checkNotNull(coder);

    this.coder = coder;

  }



  @Override

  public boolean isImmutableType() {

    return false;

  }



  @Override

  public CoderTypeSerializer<T> duplicate() {

    return new CoderTypeSerializer<>(coder);

  }



  @Override

  public T createInstance() {

    return null;

  }



  @Override

  public T copy(T t) {

    try {

      return CoderUtils.clone(coder, t);

    } catch (CoderException e) {

      throw new RuntimeException("Could not clone.", e);

    }

  }



  @Override

  public T copy(T t, T reuse) {

    return copy(t);

  }



  @Override

  public int getLength() {

    return -1;

  }



  @Override

  public void serialize(T t, DataOutputView dataOutputView) throws IOException {

    DataOutputViewWrapper outputWrapper = new DataOutputViewWrapper(dataOutputView);

    coder.encode(t, outputWrapper);

  }



  @Override

  public T deserialize(DataInputView dataInputView) throws IOException {

    try {

      DataInputViewWrapper inputWrapper = new DataInputViewWrapper(dataInputView);

      return coder.decode(inputWrapper);

    } catch (CoderException e) {

      Throwable cause = e.getCause();

      if (cause instanceof EOFException) {

        throw (EOFException) cause;

      } else {

        throw e;

      }

    }

  }



  @Override

  public T deserialize(T t, DataInputView dataInputView) throws IOException {

    return deserialize(dataInputView);

  }



  @Override

  public void copy(DataInputView dataInputView, DataOutputView dataOutputView) throws IOException {

    serialize(deserialize(dataInputView), dataOutputView);

  }



  @Override

  public boolean equals(Object o) {

    if (this == o) {

      return true;

    }

    if (o == null || getClass() != o.getClass()) {

      return false;

    }



    CoderTypeSerializer that = (CoderTypeSerializer) o;

    return coder.equals(that.coder);

  }



  @Override

  public int hashCode() {

    return coder.hashCode();

  }



  @Override

  public TypeSerializerSnapshot<T> snapshotConfiguration() {

    return new LegacySnapshot<>(this);

  }



  /** A legacy snapshot which does not care about schema compatibility. */

  public static class LegacySnapshot<T> extends TypeSerializerConfigSnapshot<T> {



    /** Needs to be public to work with {@link VersionedIOReadableWritable}. */

    public LegacySnapshot() {}



    public LegacySnapshot(CoderTypeSerializer<T> serializer) {

      setPriorSerializer(serializer);

    }



    @Override

    public int getVersion() {

      // We always return the same version

      return 1;

    }



    @Override

    public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(

        TypeSerializer<T> newSerializer) {

      // We assume compatibility because we don't have a way of checking schema compatibility

      return TypeSerializerSchemaCompatibility.compatibleAsIs();

    }

  }



  @Override

  public String toString() {

    return "CoderTypeSerializer{" + "coder=" + coder + '}';

  }

}